package cn.dmp.tools


import ch.hsr.geohash.GeoHash
import cn.dmp.utils.{BaiduGeoApi, JedisPools}
import org.apache.commons.lang.StringUtils
import org.apache.spark.sql.SQLContext
import org.apache.spark.{SparkConf, SparkContext}
import redis.clients.jedis.Jedis

/**
  * 用来抽取日志字段中的经纬度,并请求百度的api获取到商圈信息
  * 建立 经纬度=商圈 字典
  */
object ExtractLatLong2Business {
  def main(args: Array[String]): Unit = {

    if (args.length != 1) {
      println(
        """
          |cn.xiao.report.AppAnaylyseRptV2
          |参数：
          |输入路径
          |输出路径
        """.stripMargin)
      sys.exit()
    }

    val Array(logInputPath) = args

    val conf: SparkConf = new SparkConf()
      .setAppName(s"${this.getClass.getSimpleName}")
      .setMaster("local[*]")
      .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")

    val sc = new SparkContext(conf)

    val qLContext = new SQLContext(sc)

    qLContext.read.parquet(logInputPath)
      .select("lat", "long")
      //经纬度范围中国内
      .where("lat > 3 and lat < 54 and long >73 and long < 136").distinct()
      .foreachPartition(ite => {
        val jedis: Jedis = JedisPools.getJedis()
        ite.foreach(row => {
          val lat: String = row.getAs[String]("lat")
          val longs: String = row.getAs[String]("long")
          //经纬度 转 geohash编码  精度8
          val geoHashCode: String = GeoHash.withCharacterPrecision(lat.toDouble, longs.toDouble, 8).toBase32
          val busimess: String = BaiduGeoApi.Busimess(lat + "," + longs)

          if (StringUtils.isNotEmpty(busimess)) {
            jedis.set(geoHashCode, busimess)
          }
        })
        jedis.close()
      })

    sc.stop()
  }
}
